/* 
 * Copyright (c) 2019, CENTRIN.CIYUN.LTD. All rights reserved.
 */
package com.cory.boot.component.mq;

import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

/**
 * 消息队列消费者
 * @author xiongcong
 * @date 2019-07-18
 */
@Slf4j
public abstract class RocketMQConsumer implements ApplicationListener<ContextRefreshedEvent> {
    @Autowired
    private RocketMQConf rocketMQConf;
    
    public void listener(String topic) throws MQClientException {
        listener(topic, null, rocketMQConf.getConsumerGroup());
    }
    
    public void listener(String topic, String tags) throws MQClientException {
        listener(topic, tags, rocketMQConf.getConsumerGroup());
    }
    
    public void listener(String topic, String tags, String consumerGroup) throws MQClientException {
        log.info(String.format("topic=%s tags=%s", topic, tags));

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

        consumer.setNamesrvAddr(rocketMQConf.getNamesrvAddr());

        consumer.subscribe(topic, tags);
        
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 开启内部类实现监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                return RocketMQConsumer.this.dealMsg(msgs, context);
            }
        });

        consumer.start();
    }
    
    // 处理body的业务
    public abstract ConsumeConcurrentlyStatus dealMsg(List<MessageExt> msgs, ConsumeConcurrentlyContext context);
}
